package com.atguigu.bigdata.spark.core.acc;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.util.AccumulatorV2;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

public class Spark04_Acc_WordCount_JAVA {
    public static void main(String[] args) {
        // 1.创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkAcc");

        // 2. 创建sparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String>  rdd = sc.parallelize(Arrays.asList("hello", "spark", "hello"));
        MyAccumulator acc = new MyAccumulator();
        sc.sc().register(acc, "wordCountAcc");
        rdd.foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                acc.add(s);
            }
        });

        System.out.println(acc.value());

        sc.stop();
    }
}

class MyAccumulator extends AccumulatorV2<String, Map<String,Integer>> {
    private Map<String, Integer> output = new HashMap<>();

    //Returns if this accumulator is zero value or not.
    // e.g. for a counter accumulator, 0 is zero value; for a list accumulator, Nil is zero value.
    @Override
    public boolean isZero() {
        return output.isEmpty();
    }
    //复制累加器
    @Override
    public AccumulatorV2<String, Map<String, Integer>> copy() {
        MyAccumulator myAccumulator = new MyAccumulator();
        myAccumulator.output = this.output;
        return myAccumulator;
    }
    //重置累加器
    @Override
    public void reset() {
        this.output.clear();
    }
    // 获取累加器需要计算的值
    @Override
    public void add(String v) {
        int val = this.output.getOrDefault(v, 0) + 1;
        this.output.put(v,val);
    }
    // Driver合并多个累加器  output 和other 合并
    @Override
    public void merge(AccumulatorV2<String, Map<String, Integer>> other) {
        Map<String, Integer> map = other.value();
        map.forEach((k,v) -> {
            if(this.output.containsKey(k)) {
                Integer value1 = map.get(k);
                Integer value2 = this.output.get(k);
                this.output.put(k, value1 + value2);
            } else {
                this.output.put(k ,v);
            }
        });
    }
    //输出
    @Override
    public Map<String, Integer> value() {
        return this.output;
    }
}

